/*
 * Copyright (C) 2018 Baidu, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.es.api.client;

import com.google.common.collect.Lists;
import com.pingan.haofang.searchcloud.api.client.ClientFactory;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolder;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolderFactory;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author EX-SUNBIXIAO001
 * @date 2018/11/27
 * @since 1.0.0
 */
@Service
public class RestClientFactory implements ClientFactory<RestClient> {

    protected static Logger LOG = LoggerFactory.getLogger(RestClientFactory.class);

    private static final ConcurrentHashMap<String, RestClient> restClientMap = new ConcurrentHashMap<String, RestClient>();

    public static final int API_PORT = 9200;

    @Value("${es.user.authorization:}")
    private String authorization;

    @Override
    public RestClient getByName(String indexName) throws Exception {
        IndexMetadataHolder metadataHolder = IndexMetadataHolderFactory.get();
        IndexMeta meta = metadataHolder.findIndexMeta(indexName);
        if (null == meta) {
            return null;
        }
        return getByHost(meta.getHost());
    }

    @Override
    public RestClient getByHost(String host) throws Exception {
        RestClient client = restClientMap.get(host);

        if (null != client) {
            return client;
        }

        ArrayList<String> hostList = Lists.newArrayList(host.split(","));
        HttpHost[] httpHostArr = hostList.stream().map(s -> {
            String[] hostPort = s.split(":");
            return new HttpHost(hostPort[0], API_PORT, "http");

        }).toArray(HttpHost[]::new);

        Header[] defaultHeaders = new Header[]{new BasicHeader("Content-Type", "application/json")};
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        String userName = "";
        String password = "";

        if(StringUtils.isNotBlank(authorization)){
            String[] authArr = authorization.split(":");
            userName = authArr[0];
            password =authArr[1];
        }

        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName, password));
        RestClient restClient = RestClient.builder(httpHostArr)
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        return httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
                    }
                })
                .setDefaultHeaders(defaultHeaders)
                .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
                    @Override
                    public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                        return requestConfigBuilder.setConnectTimeout(5000) //连接超时（默认为1秒）
                                .setSocketTimeout(60000);//套接字超时（默认为30秒）
                    }
                }).build();

        return restClient;
    }
}
